package com.example.jtt1078_audio.callback;

import android.media.MediaCodec;
import android.media.MediaFormat;
import android.util.Log;

import androidx.annotation.NonNull;

import com.example.jtt1078_audio.constants.Constants;
import com.example.jtt1078_audio.reader.ThreadReader;
import com.example.jtt1078_audio.sender.ThreadSender;
import com.example.jtt1078_audio.encoder.audio.G711Codec;
import com.example.jtt1078_audio.encoder.audio.GetAacData;
import com.example.jtt1078_audio.interfaces.OnDataAvailable;

import java.io.InputStream;
import java.nio.ByteBuffer;
import java.util.Arrays;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.TimeUnit;

public class OnMicrophoneData implements GetAacData {
    private static final String TAG = OnMicrophoneData.class.getName();

    private static final int MTU = 1500;
    private BlockingQueue<byte[]> queue;
    private PollThread pollThread; // 数据转发
    private ThreadSender threadSender;  // 数据发送至Socket
    private ThreadReader threadReader; // Socket接收数据

    private G711Codec g711Codec;
    private int startOffset = 26;

    private ExecutorService executorService;

    // 数据发送至Socket 相关
    private String ip;
    private int port;
    private String deviceId;
    private int logicChannel;

    private OnDataAvailable onDataAvailable;

    @Override
    public void setOnDataAvailable(OnDataAvailable onDataAvailable) {
        this.onDataAvailable = onDataAvailable;
    }


    public OnMicrophoneData() {
        int cacheSize = 10 * 1024 * 1024 / MTU;
        this.queue = new LinkedBlockingDeque<>(cacheSize);
        this.g711Codec = new G711Codec();
    }

    @Override
    public void setSocketInfo(@NonNull String ip, int port, @NonNull String deviceId, int logicChannel){
        this.ip = ip;
        this.port = port;
        this.deviceId = deviceId;
        this.logicChannel = logicChannel;
    }

    @Override
    public void getAacData(@NonNull ByteBuffer aacBuffer, @NonNull MediaCodec.BufferInfo info) {
        try {
            queue.add(aacBuffer.array());
        }catch (IllegalStateException e){
            Log.e(TAG, "error, discard this frame");
            e.printStackTrace();
        }
    }

    @Override
    public void onAudioFormat(@NonNull MediaFormat mediaFormat) {

    }

    @Override
    public void start(){
        executorService = Executors.newSingleThreadExecutor();
        if(threadSender == null){
            threadSender = new ThreadSender(ip, port, deviceId, logicChannel);
            threadSender.setOnConnected(new ThreadSender.OnConnected() {
                @Override
                public void onInputStream(InputStream is) {
                    threadReader = new ThreadReader(is);
                    threadReader.setOnDataAvailable(new OnDataAvailable() {
                        @Override
                        public void onData(byte[] data) {
                            if(!Constants.playMyself){
                                if(onDataAvailable != null){
                                    onDataAvailable.onData(data);
                                }
                            }

                        }
                    });
                    threadReader.start();
                }

                @Override
                public void onSendData(byte[] data) {
                    if(Constants.playMyself){
                        if(!executorService.isShutdown()){
                            executorService.execute(()->{
                                if(onDataAvailable != null){
//                        Log.d(TAG, bytes2Hex(data));
                                    byte[] validData = Arrays.copyOfRange(data, startOffset, data.length);
//                        Log.d(TAG, bytes2Hex(validData));
                                    onDataAvailable.onData(g711Codec.decode(validData, 0, validData.length));
                                }
                            });
                        }

                    }


                }
            });
            threadSender.start();
        }
        if(pollThread == null){
            pollThread = new PollThread();
            pollThread.start();
        }

    }

    @Override
    public void stop(){
        executorService.shutdownNow();
        queue.clear();
        if(pollThread != null){
            pollThread.interrupt();
            pollThread = null;
        }
        if(threadReader != null){
            threadReader.interrupt();
            threadReader = null;
        }
        if(threadSender != null){
            threadSender.close();
            threadSender.interrupt();
            threadSender = null;
        }
    }

    class PollThread extends Thread{
        @Override
        public void run() {
            super.run();
            while (!isInterrupted()){
                try {
                   byte[] buffer = getBuffer();
                   if(buffer != null){
                       threadSender.putBuffer(buffer);
                   }
                }catch (Exception e){
                    e.printStackTrace();
                    return;
                }
            }
        }
    }

    public byte[] getBuffer(){
        try {
            return queue.poll(1, TimeUnit.SECONDS);
        } catch (InterruptedException e) {
            e.printStackTrace();
            return null;
        }
    }

    private String bytes2Hex(byte[] bytes) {
        StringBuilder sb = new StringBuilder();
        for (byte b : bytes) {
            int i = ((int) b) & 0xFF;
            String hex = Integer.toHexString(i);
            if (hex.length() < 2) {
                hex = "0"+hex;
            }
            sb.append(hex);
        }
        return sb.toString();
    }
}
